package com.shujia.flink.sql

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row

object Demo2TableApi {
  def main(args: Array[String]): Unit = {

    //创建flink的环境
    val bsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() //使用blink计划器
      .inStreamingMode() //流模式
      .build()

    /**
      * 创建flink sql 环境
      *
      */

    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(bsEnv, bsSettings)


    /**
      * 读取socket 郭建一个流
      *
      */

    val wordsDS: DataStream[String] = bsEnv.socketTextStream("master", 8888)


    /**
      * 将流转换成动态表
      *
      */

    val wordTable: Table = bsTableEnv.fromDataStream(wordsDS, $"word")

    wordTable.printSchema()


    /**
      * 在表上进行连续查询
      *
      */

    /**
      * DSL
      *
      */

    /*    val countTable: Table = wordTable
          .groupBy($"word")
          .select($"word", $"word".count().as("c"))

        countTable.printSchema()*/


    /**
      * sql
      * 需要先注册视图
      *
      */
    bsTableEnv.createTemporaryView("words", wordTable)

    val countTable: Table = bsTableEnv.sqlQuery(
      """
        |select word,count(1) from words group by word
        |
      """.stripMargin)


    /**
      * 将新的动态表转换成流
      *
      * 1、只存在追加
      * 2、有追加和更新
      *
      */

    val countDS: DataStream[(Boolean, Row)] = countTable.toRetractStream[Row]


    //打印结果
    countDS.print()

    bsEnv.execute()


  }

}
